package com.flute.haflute.jobbox.common;

import java.util.Iterator;
import java.util.Map;

import org.slf4j.*;

import com.flute.haflute.jobbox.base.InputResource;
import com.flute.haflute.jobbox.base.mapreduce.MRJobConf;
import com.flute.haflute.jobbox.base.mapreduce.MapOutputCollector;
import com.flute.haflute.jobbox.base.mapreduce.MapTaskBase;
import com.flute.haflute.jobbox.base.mapreduce.OutputCollector;
import com.flute.haflute.jobbox.communication.ChildCommunicationService;
import com.flute.haflute.jobbox.communication.Listener;
import com.flute.haflute.jobbox.communication.messages.MapperReqMessage;

/**
 * 执行map任务的调度器原型
 * 
 * 一个map任务在该调度器之下执行
 * 
 * map任务调度器执行并跟踪执行状态
 */
public class MapExecutor implements Runnable {

	private static Logger logger = LoggerFactory.getLogger(MapExecutor.class);

	private MRJobConf jobConf;
	private MapTaskBase mapTask;
	private MapperReqMessage currentRequest = null;

	// 通信平台API接口
	private ChildCommunicationService serviceFactory = null;
	private InputResource inputSource = null;

	// 需要应用特定消息交互的messager，待完善
	private Listener outputMessager = null;// should not like this ,just tmp

	private Map<String, String> inputKeyValueMap = null;
	private String mapTaskNo = "";

	@SuppressWarnings("unchecked")
	public MapExecutor(MapperReqMessage mapperRequest,
			ChildCommunicationService serviceFactory) throws Exception {

		// 初始化相关环境，实例化相关执行对象
		this.mapTaskNo = mapperRequest.getMapTaskNo();
		this.jobConf = mapperRequest.getJobConf();

		Class<? extends MapTaskBase> clazz;
		clazz = jobConf.getMapClass();
		this.mapTask = (MapTaskBase) clazz.newInstance();
		this.mapTask.configure(jobConf);

		currentRequest = mapperRequest;

		this.serviceFactory = serviceFactory;
		inputSource = (InputResource) serviceFactory.getMessager(jobConf
				.getJobId() + "_INPUT");
		inputKeyValueMap = (Map<String, String>) inputSource.getData(mapTaskNo);
		if (mapTask.hasMessager()) {
			String messagerName = jobConf.getMessagerName();
			outputMessager = (Listener) this.serviceFactory
					.getMessager(messagerName); // this should be modified
		}
	}

	public void close() throws Exception {
		if (this.mapTask != null) {
			mapTask.close();
		}
	}

	/**
	 * 真正执行并跟踪map执行的部分
	 */
	public void run() {
		long beginTime = 0;
		try {
			OutputCollector collector = new MapOutputCollector();

			beginTime = System.currentTimeMillis();

			if (inputKeyValueMap == null)
				inputKeyValueMap = currentRequest.getKeyValues();
			Iterator<String> ite = inputKeyValueMap.keySet().iterator();
			String key;
			while (ite.hasNext()) {
				key = ite.next();
				mapTask.map(collector, key, inputKeyValueMap.get(key));
			}

			if (mapTask.needFeedBack())
				inputSource.addData(collector.getVals());
			if (mapTask.hasMessager())
				outputMessager.receive(currentRequest);

			long endTime = System.currentTimeMillis();

			logger.debug("run one Map time={}", (beginTime - endTime) / 1000);
		} catch (Exception e) {
			logger.error("run() error", e);
		}
	}

	public void setCurrentRequest(MapperReqMessage currentRequest) {
		this.currentRequest = currentRequest;
	}

}
